package com.gao.hash;

import com.gao.common.ServerCache;
import com.gao.hash.algorithm.HashAlgorithm;
import com.gao.hash.algorithm.KetamaHashAlgorithm;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.TreeMap;

/**
 * 一致性哈希算法实现
 */
@Component
public class ConsistentHash {
    private static Logger logger = LoggerFactory.getLogger(ServerCache.class) ;

    /** 每个真实的缓存服务器在Hash环中存在的虚拟节点数 */
    private int VIRTUAL_NODES = 3;

    /** 所有物理服务器对应的总缓存“请求”数，负载边界中会用到 */
    private int totalLoad;

    /** 默认的 hash 算法 */
    private String defaultHashAlgorithm = "KETAMA_HASH";

    /** 服务器节点列表 */
    private Set<ServerNode> serverNodeSet;

    /**服务器节点对应的 hash key 排序树 */
    private TreeMap<Long, ServerNode> serverNodeMap = new TreeMap<>();

    private HashAlgorithm hashAlgorithm = new KetamaHashAlgorithm();

    private ConsistentHash(){}

    /**
     * 内部类持有实例对象
     */
    private static class SingletonHolder {
        private static ConsistentHash instance = new ConsistentHash();
    }

    /**
     * 获取 Hash 环实例对象
     * @return
     */
    public static ConsistentHash getInstance() {
        return SingletonHolder.instance;
    }

    /**
     * 构建哈希环，获取哈希环中的节点
     */
    public ServerNode process(List<String> serverNodes, String key) {
        serverNodeSet = new HashSet<>();
        for (String server: serverNodes) {
            serverNodeSet.add(new ServerNode(server.split(":")[0], Integer.parseInt(server.split(":")[1])));
        }

        addHash();
        return getFirstNodeValue(key);
    }

    /**
     * 添加节点到哈希环
     */
    public void addHash() {
        // 对于一个实际机器节点 T, 对应 VIRTUAL_NODES 个虚拟节点
        for (ServerNode node : serverNodeSet) {
            for (int i = 0; i < VIRTUAL_NODES; i++) {
                String key = String.format("%s_%d", node.getHost(), i);
                Long hashKey = hashAlgorithm.hash(key);
                serverNodeMap.put(hashKey, node);
            }
        }
    }

    /**
     * 根据 key 值，顺时针获取最近的一个节点
     * @param key
     * @return
     */
    public ServerNode getFirstNodeValue(String key) {
        logger.info("getFirstNodeValue param:{}", key);
        if (serverNodeMap.isEmpty()) {
            logger.error("服务获取异常！");
            return null;
        }
        // 获取 key 在这个环中的位置
        Long hashKey = hashAlgorithm.hash(key);
        Long target = hashKey;
        if (!serverNodeMap.containsKey(hashKey)) {
            // 顺时针获取大于 hashKey 的最小的键值 key
            target = serverNodeMap.ceilingKey(hashKey);
            if (serverNodeMap.ceilingKey(hashKey) == null) {
                target = serverNodeMap.firstKey();
            }
        }

        return serverNodeMap.get(target);
    }

}
